package cn.sciento.websocket.init;

import cn.sciento.core.convertor.ApplicationContextHelper;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import cn.sciento.websocket.registry.SessionRegistry;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import cn.sciento.core.redis.RedisHelper;
import cn.sciento.core.util.UUIDUtils;
import cn.sciento.websocket.config.WebSocketConfig;
import cn.sciento.websocket.redis.BrokerListenRedis;
import cn.sciento.websocket.redis.BrokerRedis;
import cn.sciento.websocket.redis.BrokerSessionRedis;
import cn.sciento.websocket.redis.SessionUserRedis;
import cn.sciento.websocket.vo.UserVO;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketSession;

@Component
public class ClientInit implements CommandLineRunner {
    @Override
    public void run(String... args) throws Exception {
        String brokerId = UUIDUtils.generateUUID();
        SessionRegistry.setBrokerId(brokerId);
        BrokerListenRedis.refreshCache(brokerId);
        ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(5, (ThreadFactory) (new BasicThreadFactory.Builder()).namingPattern("websocket-client-register").daemon(true).build());
        scheduledExecutorService.scheduleAtFixedRate(new ClientRegister(), 5L, 10L, TimeUnit.SECONDS);
        scheduledExecutorService.scheduleAtFixedRate(new Check(), 0L, 600L, TimeUnit.SECONDS);
    }

    static class ClientRegister implements Runnable {
        @Override
        public void run() {
            String brokerId = SessionRegistry.getBrokerId();
            BrokerRedis.refreshCache(brokerId);
            List<String> brokerList = BrokerListenRedis.getCache();
            if (!brokerList.contains(brokerId)) {
                BrokerListenRedis.refreshCache(brokerId);
            }
            brokerList.forEach(item -> {
                if (!BrokerRedis.isAlive(item)) {
                    BrokerListenRedis.clearRedisCache(item);
                    BrokerSessionRedis.clearRedisCacheByBrokerId(item);
                }
            });
        }
    }

    static class Check implements Runnable {
        private RedisHelper redisHelper = ApplicationContextHelper.getContext().getBean(RedisHelper.class);

        private WebSocketConfig webSocketConfig = ApplicationContextHelper.getContext().getBean(WebSocketConfig.class);

        @Override
        public void run() {
            List<UserVO> userList;
            String brokerId = SessionRegistry.getBrokerId();
            List<String> liveBrokerList = BrokerListenRedis.getCache();
            int page = 0;
            int size = 500;
            do {
                userList = SessionUserRedis.getCache(page, size);
                for (UserVO item : userList) {
                    String sessionId = item.getSessionId();
                    if (StringUtils.isBlank(item.getBrokerId()) || !liveBrokerList.contains(item.getBrokerId())) {
                        SessionUserRedis.deleteCache(item);
                        SessionRegistry.removeSession(sessionId);
                        continue;
                    }
                    if (!Objects.equals(item.getBrokerId(), brokerId)) {
                        continue;
                    }
                    if (checkAccessToken(item.getAccessToken())) {
                        clear(item);
                        continue;
                    }
                    WebSocketSession webSocketSession = SessionRegistry.getSession(sessionId);
                    if (webSocketSession == null || !webSocketSession.isOpen()) {
                        clear(item);
                    }
                }
                page++;
            } while (userList.size() >= size);
        }

        private void clear(UserVO user) {
            String sessionId = user.getSessionId();
            Long userId = SessionRegistry.getUser(sessionId);
            if (userId != null) {
                BrokerSessionRedis.deleteCache(user.getBrokerId(), userId, sessionId);
            }
            SessionUserRedis.deleteCache(user);
            SessionRegistry.removeSession(sessionId);
        }

        private boolean checkAccessToken(String accessToken) {
            this.redisHelper.setCurrentDatabase(this.webSocketConfig.getOauthRedisDb());
            String key = "access_token:access:" + accessToken;
            boolean unable = true;
            if (this.redisHelper.hasKey(key)) {
                unable = false;
            }
            this.redisHelper.clearCurrentDatabase();
            return unable;
        }
    }
}
